package com.lm.flink;

import com.lm.flink.component.FlinkKafkaToRedisJob;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import java.util.Properties;

/**
 * Fastjson   是一个快速、功能强大的 JSON 序列化和反序列化库
 */
@SpringBootApplication
public class FlinkApplication implements  CommandLineRunner {

    @Autowired
    private FlinkKafkaToRedisJob job;

    public static void main(String[] args) {
        SpringApplication.run(FlinkApplication.class, args);
    }

    @Override
    public void run(String... args) throws Exception {
        job.runJob();
    }

//    @Override
//    public void run(String... args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//
//        // 生成数据流（模拟数据）
//        DataStream<String> textStream = env.fromElements(
//                "hello flink",
//                "hello spring",
//                "flink spring boot"
//        );
//
//        // 词频统计
//        DataStream<Tuple2<String, Integer>> wordCounts = textStream
//                .flatMap((String line, Collector<Tuple2<String, Integer>> out) -> {
//                    for (String word : line.split(" ")) {
//                        out.collect(new Tuple2<>(word, 1));
//                    }
//                })
//                .returns(Types.TUPLE(Types.STRING, Types.INT))
//                .keyBy(t -> t.f0)
//                .sum(1);
//
//        // 打印到控制台
//        wordCounts.print();
//
//        // 启动 Flink 任务
//        env.execute("Flink WordCount with Spring Boot");
//    }
}